package cn.doitedu.hbase.day02;

import cn.doitedu.hbase.day01.ConnectionUtils;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.util.Bytes;

import java.util.List;
import java.util.Map;
import java.util.NavigableMap;
import java.util.Set;

public class GetDemo2 {

    public static void main(String[] args) throws Exception {

        //获取连接
        Connection connection = ConnectionUtils.getConnection();

        //获取对应的表对象
        Table table = connection.getTable(TableName.valueOf("tb_users"));

        //获取指定列族的全部key-value
        //getAllKeyValuesOnFamily(table);

        getCellAllKeyValue(table);



        table.close();
        connection.close();

    }

    /**
     * 查询指定列族下的全部key-value
     */
    private static void getAllKeyValuesOnFamily(Table table) throws Exception{

        Get get = new Get(Bytes.toBytes("u001"));
        //添加查询条件(指定列族)
        get.addFamily(Bytes.toBytes("f1"));
        get.addFamily(Bytes.toBytes("f2"));

        Result result = table.get(get);

        NavigableMap<byte[], byte[]> familyMap = result.getFamilyMap(Bytes.toBytes("f1"));

        Set<Map.Entry<byte[], byte[]>> entries = familyMap.entrySet();
        for (Map.Entry<byte[], byte[]> entry : entries) {

            String key = Bytes.toString(entry.getKey());
            String value = Bytes.toString(entry.getValue());
            System.out.println(key + " : " + value);
        }


    }


    private static void getCellAllKeyValue(Table table) throws Exception{

        Get get = new Get(Bytes.toBytes("u006"));
        //可以获取多个版本（默认只获取最新版本）
        //get.setMaxVersions(3);
        get.readVersions(5);

        Result result = table.get(get);

        //如果version是1，那么就会覆盖
        List<Cell> columnCells = result.getColumnCells(Bytes.toBytes("f1"), Bytes.toBytes("name"));

        //返回最新版本的数据
        //Cell cell = result.getColumnLatestCell(Bytes.toBytes("f1"), Bytes.toBytes("name"));

        for (Cell cell : columnCells) {

            //通过下面的方式获取，会有乱码
//            byte[] familyArray = cell.getFamilyArray();
//            String family = Bytes.toString(familyArray);
//
//            byte[] qualifierArray = cell.getQualifierArray();
//            String qualifier = Bytes.toString(qualifierArray);
//
//            byte[] valueArray = cell.getValueArray();
//            String value = Bytes.toString(valueArray);


            byte[] familyArray = CellUtil.cloneFamily(cell);
            String family = Bytes.toString(familyArray);

            byte[] qualifierArray = CellUtil.cloneQualifier(cell);
            String qualifier = Bytes.toString(qualifierArray);

            byte[] valueArray = CellUtil.cloneValue(cell);
            String value = Bytes.toString(valueArray);

            System.out.println(family + " , " + qualifier + " , " + value);


        }


    }

}
